/**     
 * @Title: AbstractSubscriber.java   
 * @Package me.xxy.mq   
 * @Description: TODO
 * @author weiwei 
 * @date 2018年5月31日 下午4:15:28   
 * @version V1.0     
 */
package me.xxy.mq;

import javax.annotation.PostConstruct;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.annotation.Autowired;

import redis.clients.jedis.JedisPubSub;

/**
 * @ClassName: AbstractSubscriber
 * @Description: TODO
 * @author weiwei
 * @date 2018年5月31日 下午4:15:28
 * 
 */
public abstract class AbstractSubscriber extends JedisPubSub implements Runnable {

	private static final Log LOG = LogFactory.getLog(AbstractSubscriber.class);

	@Autowired
	protected MessageTrunk messageTrunk;

	public AbstractSubscriber(String... channels) {
		super();
		this.channels = channels;
	}

	@PostConstruct
	public void start() {
		// 启动监听
		new Thread(this).start();
	}

	/** 频道 **/
	private String[] channels;

	@Override
	public void run() {
		while (true) {
			listen();
		}
	}

	/**
	 * @Title: listen
	 * @Description: 监听消息
	 * @Reutrn void
	 */
	public void listen() {
		try {
			if (null == channels || channels.length <= 0) {
				LOG.warn("Channels cannot be empty");
				return;
			}
			LOG.debug(this);
			LOG.debug("channels >> " + channels);
			messageTrunk.subscribed(this, channels);
		} catch (Exception ex) {
			LOG.error("消息总线发生异常", ex);
		}
	}

	// 取得订阅的消息后的处理
	@Override
	public abstract void onMessage(String channel, String message);

	// 初始化订阅时候的处理
	@Override
	public void onSubscribe(String channel, int subscribedChannels) {
		LOG.error(String.format("Subscriber subscribe redis channel success, channel %s, subscribedChannels %d", channel, subscribedChannels));
		super.onSubscribe(channel, subscribedChannels);
	}

	// 取得按表达式的方式订阅的消息后的处理
	@Override
	public void onPMessage(String pattern, String channel, String message) {
		LOG.error(String.format("Subscriber PMessage redis channel success, pattern %s, channel %s, message %d", pattern, channel, message));
		super.onPMessage(pattern, channel, message);
	}

	// 取消订阅时候的处理
	@Override
	public void onUnsubscribe(String channel, int subscribedChannels) {
		LOG.error(String.format("Subscriber unsubscribe redis channel success, channel %s, subscribedChannels %d", channel, subscribedChannels));
		super.onUnsubscribe(channel, subscribedChannels);
	}

	// 取消按表达式的方式订阅时候的处理
	@Override
	public void onPUnsubscribe(String pattern, int subscribedChannels) {
		LOG.error(String.format("Subscriber PUnsubscribe redis channel success, pattern %s, subscribedChannels %d", pattern, subscribedChannels));
		super.onPUnsubscribe(pattern, subscribedChannels);
	}

	// 初始化按表达式的方式订阅时候的处理
	@Override
	public void onPSubscribe(String pattern, int subscribedChannels) {
		LOG.error(String.format("Subscriber PSubscribe redis channel success, pattern %s, subscribedChannels %d", pattern, subscribedChannels));
		super.onPSubscribe(pattern, subscribedChannels);
	}
}
